package gg.zsw.datasource.demo.kafka;

import gg.zsw.dataframework.decode.DecodeChain;
import gg.zsw.dataframework.decode.DecodeHandler;
import gg.zsw.dataframework.pubsub.AbstractObserver;
import org.springframework.context.annotation.Bean;


/**
 * 提供数据管理器，观察者是抽象的
 *
 * @param <Observer> 收听数据的对象，Message是程序所能使用的最终数据类型
 * @author zhou.sw
 * @Date 2020/7/1 10:51
 */
public abstract class MessageManager<Observer extends AbstractObserver<Message>> {


    /**
     * 消息主题类。直接与kafka沟通的，分发数据到MessageSubject中
     */
    protected MessageTopic<MessageSubject<Observer, Message>> messageTopic;

    public MessageManager() {
        this.messageTopic = new MessageTopic<>();
    }

    /**
     * 获取指定Subject，并自动注册到messageTopic中去。
     * 使用方并不想知道底层怎么把数据分发过来的。只想知道找哪个MessageSubject可以找到给他数据主题。
     * 这样通过 MessageSubject
     * <p>
     * 注意: 容易造成，只注册，不注销的情况
     *
     * @param key
     * @return
     */
    public MessageSubject<Observer, Message> getMessageSubject(String key) {
        return messageTopic.getObserverAutoRegisterIfAbsent(key, this::createMessageSubject);
    }

    /**
     * 创建方法
     * 需要去重写这个方法
     * Manager只是隔离一下数据源头，作为使用方，接入什么样的观察者，使用什么样的数据，管不了呢。
     *
     * @return
     */
    public MessageSubject<Observer, Message> createMessageSubject(String key) {

        //注意: 无状态的处理链，其实直接static就行
        DecodeChain decodeChain = new DecodeChain();
        decodeChain.addDecodeHandler(new ZswDecodeHandler());

        MessageSubject<Observer, Message> messageSubject =
                new MessageSubject<Observer, Message>(decodeChain) {
                    @Override
                    public long getNotifyDataTime(Message msg) {
                        return msg.getTime();
                    }

                    @Override
                    public void nonObserverCallBack() {
                        messageTopic.removeObserver(key);
                    }
                };
        return messageSubject;
    }

    /**
     * 通过Bean的方式把messageTopic暴露出去，其实也可以直接暴露这个变量。
     * MessageManager.get也是阔以哒
     *
     * @return
     */
    @Bean
    protected MessageTopic<MessageSubject<Observer, Message>> messageTopic() {
        return this.messageTopic;
    }


    //  ===============================================================================
    //  构建相关DecodeHandler

    class ZswDecodeHandler implements DecodeHandler<String, Message> {
        // TODO: 2020/6/30 其实要用dao去丰富数据，这里先随便搞搞

        @Override
        public Message decode(String key) {

            String[] split = key.split(":");
            long time = Long.parseLong(split[4]);
            // TODO: 2020/6/30 填充指定时间的数据。
            Message msg = new Message();
            msg.setTime(time);
            msg.setData(key);
            return msg;
        }
    }
}
